#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <limits.h>
#include <dirent.h>
#include <sys/time.h>
#include <fcntl.h>
#include <sys/stat.h>
#include <sys/socket.h>

#include "data_type.h"
#include "cube.h"
#include "cube_define.h"
#include "cube_record.h"

#include "broadcast.h"

int broadcast_init(void * sub_proc,void * para)
{
	int ret;
	// add youself's plugin init func here
	return 0;
}

int broadcast_start(void * sub_proc,void * para)
{
	int ret;
	void * recv_msg;
	int i;
	int type;
	int subtype;

	while(1)
	{
		usleep(time_val.tv_usec);
		ret=ex_module_recvmsg(sub_proc,&recv_msg);
		if(ret<0)
			continue;
		if(recv_msg==NULL)
			continue;
		type=message_get_type(recv_msg);
		subtype=message_get_subtype(recv_msg);
		if(!memdb_find_recordtype(type,subtype))
		{
			printf("message format (%d %d) is not registered!\n",
				message_get_type(recv_msg),message_get_subtype(recv_msg));
			ret=proc_broadcast_transrecord(sub_proc,recv_msg);
			continue;
		}
		if((type==TYPE(MESSAGE))||(subtype == SUBTYPE(MESSAGE,CONN_ACKI)))
		{
			ret=proc_broadcast_addclient(sub_proc,recv_msg);
		}
		else if((type==TYPE(GENERAL_RETURN))||(subtype == SUBTYPE(GENERAL_RETURN,STRING)))
		{
			ret=proc_broadcast_notice(sub_proc,recv_msg);
		}
       		else
       		{
			ret=proc_broadcast_transrecord(sub_proc,recv_msg);
       		}
	}
	return 0;
}

int proc_broadcast_addclient(void * sub_proc,void * recv_msg)
{
	int ret;

	RECORD(MESSAGE,INSTANCE_INFO) * instance_info;
	RECORD(GENERAL_RETURN,STRING) * item_select;
	RECORD(MESSAGE,CONN_ACKI) * conn_acki;
	RECORD(GENERAL_RETURN,UUID) * instance_uuid;
	
	DB_RECORD * db_record;
	MSG_EXPAND * msg_expand;

    ret=message_get_record(recv_msg,&conn_acki,0);
    if(ret<0)
         return ret;
    if(conn_acki==NULL)
         return -EINVAL;
	
	instance_info = Talloc0(sizeof(*instance_info));
	if(instance_info == NULL)
			return -ENOMEM;
	BYTE uuid[DIGEST_SIZE];

	 Strncpy(instance_info->proc_name,conn_acki->client_proc,DIGEST_SIZE);
	 Memcpy(instance_info->node_uuid,conn_acki->uuid,DIGEST_SIZE);
     
	 memdb_store(instance_info,TYPE_PAIR(MESSAGE,INSTANCE_INFO),instance_info->proc_name);
    
     // memdb_comp_record_uuid(instance_info,TYPE_PAIR(MESSAGE,INSTANCE_INFO),uuid);
	//db_record=memdb_find(uuid,TYPE_PAIR(MESSAGE,INSTANCE_INFO));

    	item_select = ex_module_getpointer(sub_proc);
	if(item_select!=NULL)
	{
//		if(Strcmp(item_select->name,"item_select")==0)
//		{
        	instance_uuid=Talloc0(sizeof(*instance_uuid));
        	if(instance_uuid==NULL)
            		return -ENOMEM;
		instance_uuid->name = dup_str("broaddcast",0);
		comp_proc_uuid(instance_info->node_uuid,instance_info->proc_name,instance_uuid->return_value);

		void * send_msg = message_create(TYPE_PAIR(GENERAL_RETURN,STRING),NULL);
		if(send_msg==NULL)
			return -ENOMEM;
		message_add_record(send_msg,item_select);
        	ret=message_add_expand_data(send_msg,TYPE_PAIR(GENERAL_RETURN,UUID),instance_uuid);
        	if(ret<0)
            		return ret;
        	ex_module_sendmsg(sub_proc,send_msg);
//		}
	}
	return 0;
}

int proc_broadcast_transrecord(void * sub_proc,void * recv_msg)
{
	int ret;

	RECORD(MESSAGE,INSTANCE_INFO) * instance_info;
	RECORD(GENERAL_RETURN,UUID) * instance_uuid;
	
	DB_RECORD * db_record;
	MSG_EXPAND * msg_expand;
   	void * new_msg;

	db_record=memdb_get_first(TYPE_PAIR(MESSAGE,INSTANCE_INFO));
   	if(db_record==NULL)
   	{
        	 print_cubeerr("no client in memdb!");
           	return -EINVAL;
    	}

    	while(db_record!=NULL)
    	{
	    instance_info=db_record->record;
            if(instance_info==NULL)
            	return -EINVAL;

        instance_uuid=Talloc0(sizeof(*instance_uuid));
        if(instance_uuid==NULL)
            return -ENOMEM;

	instance_uuid->name = dup_str("broaddcast",0);
	comp_proc_uuid(instance_info->node_uuid,instance_info->proc_name,instance_uuid->return_value);
    
        new_msg=message_clone(recv_msg);
        if(new_msg==NULL)
                return -EINVAL;
        ret=message_add_expand_data(new_msg,TYPE_PAIR(GENERAL_RETURN,UUID),instance_uuid);
        if(ret<0)
                return ret;
        ex_module_sendmsg(sub_proc,new_msg);
	    db_record=memdb_get_next(TYPE_PAIR(MESSAGE,INSTANCE_INFO));
    }
	return 0;
}


int proc_broadcast_notice(void * sub_proc,void * recv_msg)
{
	int ret;

	RECORD(GENERAL_RETURN,STRING) * msg_notice;
	RECORD(MESSAGE,INSTANCE_INFO) * instance_info;
	RECORD(GENERAL_RETURN,UUID) * instance_uuid;
	
	DB_RECORD * db_record;
	MSG_EXPAND * msg_expand;
	void * new_msg;

    	ret=message_get_record(recv_msg,&msg_notice,0);
    	if(ret<0)
         return ret;
    	if(msg_notice==NULL)
         	return -EINVAL;

	print_cubeaudit("verify_proxy: select item %s",msg_notice->name);
		ex_module_setpointer(sub_proc,msg_notice);
	/*
	if(Strcmp(msg_notice->name,"item_select")==0)
	{
		print_cubeaudit("verify_proxy: select item %s",msg_notice->return_value);
		ex_module_setpointer(sub_proc,msg_notice);
		return 0;
	}
	*/

//	ret=proc_broadcast_transrecord(sub_proc,recv_msg);
}

